package prictice.lesson;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.io.*;
import java.util.Map;

/**
 * @Package prictice.lesson
 * @Description:
 * @Author elwyn
 * @Date 2017/9/15 0:10
 * @Email elonyong@163.com
 */
public class MySpout implements IRichSpout {
    private FileInputStream fileInputStream;
    private BufferedReader bufferedReader;
    private InputStreamReader inputStreamReader;
    private SpoutOutputCollector spoutOutputCollector = null;

    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        try {
            this.spoutOutputCollector = spoutOutputCollector;
            this.fileInputStream = new FileInputStream("track.log");
            this.inputStreamReader = new InputStreamReader(fileInputStream, "utf-8");
            this.bufferedReader = new BufferedReader(inputStreamReader);
        } catch (FileNotFoundException | UnsupportedEncodingException e) {
            e.printStackTrace();
        }
    }


    String str = null;

    @Override
    public void nextTuple() {
        try {
            while ((str = this.bufferedReader.readLine()) != null) {
                spoutOutputCollector.emit(new Values(str));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void ack(Object o) {

    }

    @Override
    public void fail(Object o) {

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("log"));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }


    @Override
    public void close() {

    }

    @Override
    public void activate() {

    }

    @Override
    public void deactivate() {

    }
}
